package com.neusoft.databus.policy.config;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterReturning;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.annotation.Pointcut;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.google.common.eventbus.AsyncEventBus;
import com.neusoft.bizcore.common.bean.MessageBean;
import com.neusoft.bizcore.web.utils.JsonUtils;
import com.neusoft.bizcore.web.utils.id.BaseIdGenerator;
import com.neusoft.bizcore.web.utils.id.DefaultIdGenerator;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.utils.DefaultThreadFactory;
import com.neusoft.databus.policy.constants.DatabusPolicyConstant;

@Aspect
@Configuration
public class DatabusPolicyConfiguration {
    @Value("${databus.policy.policyEventDispatcherMinThreadSize: 16}")
    private int policyEventDispatcherMinThreadSize;
    @Value("${databus.policy.policyEventDispatcherMaxThreadSize: 32}")
    private int policyEventDispatcherMaxThreadSize;
    // 策略编码前缀
    private static final String POLICY_CODE_PREFIX = "PL_";

    @Bean
    public BaseIdGenerator policyCodeGenerator() {
        return new DefaultIdGenerator(0, DatabusPolicyConfiguration.POLICY_CODE_PREFIX);
    }

    @Bean
    public AsyncEventBus policyEventBus() {
        final ExecutorService executorService = new ThreadPoolExecutor(
                Integer.valueOf(this.policyEventDispatcherMinThreadSize),
                Integer.valueOf(this.policyEventDispatcherMaxThreadSize),
                120L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
                new DefaultThreadFactory(DatabusPolicyConstant.POLICY_EVENT_EVENTBUS_ID),
                new ThreadPoolExecutor.CallerRunsPolicy());
        return new AsyncEventBus(DatabusPolicyConstant.POLICY_EVENT_EVENTBUS_ID, executorService);
    }

    /**
     * 定义切点Pointcut <br>
     * 第一个*号：表示返回类型， *号表示所有的类型 <br>
     * 第二个*号：表示类名，*号表示所有的类 <br>
     * 第三个*号：表示方法名，*号表示所有的方法 <br>
     * 后面括弧里面表示方法的参数，两个句点表示任何参数
     */
    @Pointcut("execution(* com.neusoft.databus.policy.listener..*.processMessage(..))")
    public void executionService() {
    }

    @Before(value = "executionService()")
    public void doBefore(final JoinPoint joinPoint) {
        final Object arg = joinPoint.getArgs()[0];
        if (arg instanceof MessageBean) {
            final MessageBean message = (MessageBean) arg;
            final CommandBean commandBean = JsonUtils.jsonToPojo(message.getMessage(), CommandBean.class);
            final String traceId = commandBean.generateTraceId();
            MDC.put("trace", traceId);
        }
    }

    @AfterReturning(pointcut = "executionService()", returning = "returnValue")
    public void doAfterReturning(final JoinPoint joinPoint, final Object returnValue) {
        MDC.clear();
    }
}
